package com.audaque.springboot.foshanupload.web.amqpdemo.listener;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @author zgb
 * @desc ...
 * @date 2024-01-04 01:11:31
 */
@Slf4j
@Component
public class SpringRabbitListener {
  /**
   * s: 简单队列：只有一个消费者
   */
  //方式1

  /**
   * queuesToDeclare：The queues for this listener. If there is a org.springframework.amqp.rabbit.core.RabbitAdmin
   * in the application context, the queue will be declared on the broker with default binding
   * (default exchange with the queue name as the routing key).
   */
  @RabbitListener(queuesToDeclare = {@Queue("simple.queue")})
  //方式2
  //@RabbitListener(queues = "mall.order.cancel")//队列名
  public void listenSimpleQueueMessage(String msg) throws InterruptedException {
    System.out.println("spring 消费者接收到消息：【" + msg + "】");
  }

  /**
   * e: 简单队列：只有一个消费者
   */

  /**
   * s: 工作队列：发送到相同的队列，多个消费者公平消费或竞争消费
   */
  @RabbitListener(queuesToDeclare = @Queue("workQueue")) // queuesToDeclare 自动声明队列
  public void holloWordListener(String message) throws InterruptedException {
    Thread.sleep(200);
    System.out.println("message1 = " + message);
  }

  @RabbitListener(queuesToDeclare = @Queue("workQueue")) // queuesToDeclare 自动声明队列
  public void holloWordListener1(String message) throws InterruptedException {
    Thread.sleep(400);
    System.out.println("message2 = " + message);
  }
  /**
   * e: 工作队列：发送到相同的队列，多个消费者公平分配消费或竞争消费
   */


  /**
   * s: 广播队列：发送到相同的交换器，多个消费者都能消费
   */
  //指定队列，可以接收缓存到队列里的消息
  @RabbitListener(bindings = {
  @QueueBinding(
  value = @Queue(
  value = "test",
  durable = "true",
  arguments = @Argument(name = "x-queue-mode", value = "lazy")
  ),
  exchange = @Exchange(
  value = "fanoutTest",
  type = ExchangeTypes.FANOUT)
  )
  })
  public void reveivel(String message) {
    System.out.println("message在我这里发送了短信 = " + message);
  }

  //不显式声明则会自动生成名字
  @RabbitListener(bindings = {@QueueBinding(value = @Queue,
  exchange = @Exchange(value = "fanoutTest", type = ExchangeTypes.FANOUT))})
  public void reveivel2(String message) {
    System.out.println("message1在我这里发送了 邮件 = " + message);
  }
  /**
   * e: 广播队列：发送到相同的交换器，多个消费者都能消费
   */


  /**
   * s: 路由模式：一个vhost内的队列名称唯一，一个队列可以被多个交换器+路由键绑定，发送消息时用交换器+路由键来寻址队列
   */
  @RabbitListener(bindings = {@QueueBinding(value = @Queue(value = "test1", durable = "true", arguments = @Argument(name = "x-queue-mode", value = "lazy")),
  exchange = @Exchange(value = "direstTest", type = ExchangeTypes.DIRECT),
  key = {"info", "error"})})
  public void receivel(String message) {
    System.out.println("message = " + message);
  }

  @RabbitListener(bindings = {@QueueBinding(value = @Queue,
  exchange = @Exchange(value = "direstTest", type = ExchangeTypes.DIRECT),
  key = {"error"})})
  public void receivel1(String message) {
    System.out.println("message1 = " + message);
  }

  /**
   * e: 路由模式：一个vhost内的队列明湖曾唯一，一个队列可以被多个交换器+路由键绑定，发送消息时用交换器+路由键来寻址队列
   */

  /**
   * s: 主题模式: 提供key模糊查询
   */
  @RabbitListener(bindings = {@QueueBinding(value = @Queue(value = "test2", durable = "true", arguments = @Argument(name = "x-queue-mode", value = "lazy")),
  exchange = @Exchange(name = "topicList", type = ExchangeTypes.TOPIC), key = {"user.save", "user.*"})})
  public void recevicel(String message) {
    System.out.println("message = " + message);
  }


  @RabbitListener(bindings = {@QueueBinding(value = @Queue,
  exchange = @Exchange(name = "topicList", type = ExchangeTypes.TOPIC), key = {"order.#", "user.*"})})
  public void recevicel1(String message) {
    System.out.println("message1 = " + message);
  }

  /**
   * e: 主题模式：提供key模糊查询
   */


  @RabbitListener(bindings = @QueueBinding(
  value = @Queue(name = "delay.queue", durable = "true"),
  exchange = @Exchange(name = "delay.direct", delayed = "true"),
  key = "delay"
  ))
  public void listenDelayMessage(String msg) {
    log.debug("接收到delay.queue的延迟消息：{}", msg);
  }


  @RabbitListener(queues = "delay_queue")
  public void receive(String msg) {
    System.out.println("接收到的消息:"+msg);
  }


  /**
   * binding已经在config文件声明，直接指向队列
   * @param message
   * @param headers
   * @param channel
   */
  @RabbitListener(queues = "direct_boot_queue")
  public void listener(String message, @Headers Map<String, Object> headers, Channel channel) {
    log.debug("Consumer receive new message: {}", message);
    final String exchange = (String) headers.get(AmqpHeaders.RECEIVED_EXCHANGE);
    final String queue = (String) headers.get(AmqpHeaders.CONSUMER_QUEUE);
    final String routingKey = (String) headers.get(AmqpHeaders.RECEIVED_ROUTING_KEY);
    final Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);


    try {
      // 可以在此做一些业务操作
      // 也可以将此消费者测试用例封装起来，用java.util.function.Consumer作为参数传入，在此进行consumer.accept(messageBodyList)，以确保业务操作成功之后ack
      //处理业务

      //处理业务后应答

      /**
       * 无异常就确认消息
       * basicAck(long deliveryTag, boolean multiple)
       * deliveryTag:该消息的index <br>
       * multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息
       */
      channel.basicAck(deliveryTag, false);
    } catch (Exception e) {
      log.error("Abnormal consumption!", e);

      try {
        /**
         * 有异常就绝收消息
         * basicNack(long deliveryTag, boolean multiple, boolean requeue)
         * requeue:true为将消息重返当前消息队列,还可以重新发送给消费者;
         *         false:将消息丢弃
         */
        channel.basicNack(deliveryTag,false,true);
      } catch (Exception ex) {
        log.error("Reject message exception!", ex);
      }
    }




  }


}

